package itemCF.step3;


import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * @author legolas
 * @date 2020/3/13 下午7:06
 */

public class Reducer3 extends Reducer<Text, Text, Text, Text> {

    /**
     * 需要实现的reduce函数
     * 将map分组后的k2，v2s进行聚合处理
     *
     * @param k2
     * @param v2s
     * @param context
     * @throws IOException
     * @throws InterruptedException
     */

    @Override
    protected void reduce(Text k2, Iterable<Text> v2s, Context context) throws IOException, InterruptedException {
        StringBuilder sb = new StringBuilder();
        for (Text v2 : v2s) {
            sb.append(v2).append(",");
        }
        String line = null;
        if (sb.toString().endsWith(",")) {
            line = sb.toString().substring(0, sb.length() - 1);
        }
        Text k3 = new Text();
        k3.set(k2);
        Text v3 = new Text();
        assert line != null;//等同与if(line!=null)
        v3.set(line);
        /**
         * 最终输出：
         * A    1_2.0    2_10.0     3_0     4_3.0       5_0     6_5.0
         * B    1_0    2_3.0     3_0    ……
         * C    1_5.0    2_0     3_15.0     ……
         */

        context.write(k3, v3);

    }
}
